package spring.cloud.eurekaclientproduct.service.impl;

import io.github.resilience4j.bulkhead.Bulkhead;
import io.github.resilience4j.bulkhead.BulkheadConfig;
import io.github.resilience4j.bulkhead.BulkheadRegistry;
import io.github.resilience4j.cache.Cache;
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
import io.github.resilience4j.circuitbreaker.CircuitBreaker.State;
import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig;
import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry;
import io.github.resilience4j.ratelimiter.RateLimiter;
import io.github.resilience4j.ratelimiter.RateLimiterRegistry;
import io.github.resilience4j.retry.Retry;
import io.github.resilience4j.retry.RetryConfig;
import io.github.resilience4j.retry.RetryRegistry;
import io.github.resilience4j.timelimiter.TimeLimiter;
import io.github.resilience4j.timelimiter.TimeLimiterConfig;
import io.vavr.CheckedFunction0;
import io.vavr.CheckedFunction1;
import io.vavr.control.Try;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.validation.Errors;
import org.springframework.web.client.HttpServerErrorException;
import org.springframework.web.client.RestTemplate;
import spring.cloud.eurekaclientproduct.service.R4jFacade;
import com.spring.cloud.commonutils.vo.UserInfo;
import com.spring.cloud.commonutils.vo.ResultMessage;
import javax.cache.CacheManager;
import javax.cache.Caching;
import javax.cache.configuration.MutableConfiguration;
import javax.cache.spi.CachingProvider;
import javax.xml.transform.Result;
import java.time.Duration;
import java.util.concurrent.*;
import java.util.function.Supplier;

/**
 * Resilience4j 断路器业务实现类
 * @author Administrator
 */
@Service
public class R4jFacadeImpl implements R4jFacade {
//断路器
    @Autowired
    private CircuitBreakerRegistry circuitBreakerRegistry;

//    @Autowired
//    private CircuitBreakerConfig circuitBreakerConfig;

    //限速器
    @Autowired
    private RateLimiterRegistry rateLimiterRegistry;

    //舱壁注册机
    @Autowired
    private BulkheadRegistry bulkheadRegistry;
    //舱壁配置
//    @Autowired
//    private BulkheadConfig bulkheadConfig;

    //重试注册机
    @Autowired
    private RetryRegistry retryRegistry;
    //重试配置
    @Autowired
    private RetryConfig retryConfig;

    //时间限制器配置
    @Autowired
    private TimeLimiterConfig timeLimiterConfig;

    //restful模板
    @Autowired
    private RestTemplate restTemplate;

    //错误结果ID
    private static final  long ERROR_ID = Long.MAX_VALUE;

    @Override
    public ResultMessage exp(String msg) {

        //根据配置 创建断路器,exp 为键
        CircuitBreaker circuitBreaker = circuitBreakerRegistry.circuitBreaker("exp");


        //模拟异常场景事件
        //在1000纳秒后,以运行时异常(RuntimeException) ,模拟一次断路器错误情况
//        circuitBreaker.onError(1000, TimeUnit.MICROSECONDS,new RuntimeException());
        //在1000纳秒后,模拟一次断路器执行成功
//        circuitBreaker.onSuccess(1000,TimeUnit.MICROSECONDS);


        //手动操作断路器
        //重置断路器,同时删除度量数据
//        circuitBreaker.reset();
//        将断路器状态修改为CLOSED
//        circuitBreaker.transitionToClosedState();


        //获取度量信息
//        CircuitBreaker.Metrics metrics = circuitBreaker.getMetrics();
//        //获取环形位缓冲区中记录的调用失败比例
//        float failureRate = metrics.getFailureRate();
//        //获取环形位缓冲区中记录的调用次数
//        int numberOfBufferedCalls = metrics.getNumberOfBufferedCalls();
//        //获取环形位缓冲区中记录的调用失败次数
//        int numberOfFailedCalls = metrics.getNumberOfFailedCalls();
//        //获取环形位缓冲区中记录的调用成功次数
//        int numberOfSuccessfulCalls = metrics.getNumberOfSuccessfulCalls();

        //2 监听事件
        //获取事件发布者
        circuitBreaker.getEventPublisher()
                //调用不允许调用事件,例如,断路器打开,不允许请求
                .onCallNotPermitted(env->{
                    System.out.println("[onCallNotPermitted]"+env.getEventType().name());
                })
                //调用错误事件
                .onError(env->{
                    System.out.println("[onError]"+env.getEventType().name());
                })
                //调用成功事件
                .onSuccess(env->{
                    System.out.println("[onSuccess]"+env.getEventType().name());
                })
                //断路器状态改变事件
                .onStateTransition(env->{
                    System.out.println("[onStateTransition]"+env.getEventType().name());
                })
                //忽略错误事件
                .onIgnoredError(env->{
                    System.out.println("[onIgnoredError]"+env.getEventType().name());
                })
                //重置断路器事件,关于重置后续会谈及
                .onReset(env -> {
                    System.out.println("[onReset]"+env.getEventType().name());
                })
                //注册消费者监听所有的事件
                .onEvent(env -> {
                    System.out.println("[onEvent]"+env.getEventType().name());
                });


        //1是代码正常执行
        //描述事件,准备发送
        CheckedFunction0<ResultMessage> decoratedSupplier = CircuitBreaker.decorateCheckedSupplier(circuitBreaker,()->{
            String url = "http://eureka-client-user/hystrix/exp/{msg}";
            System.out.println("发送消息["+msg +"]");
            return restTemplate.getForObject(url,ResultMessage.class,msg);
        });



        //获取断路器的状态
        State state =  circuitBreaker.getState();
        System.out.println("断路器状态:["+state.name()+"]");
        //发送事件
        Try<ResultMessage> result = Try.of(decoratedSupplier)
                //3 如果发生异常,执行降级方法
                .recover(ex->{
                    System.out.println("进行方法降级");
                    return new ResultMessage(false,ex.getMessage());
                });
        return result.get();
    }


    /**
     * 限速器的使用
     * @param id
     * @return
     */
    @Override
    public UserInfo getUser(Long id){
        //获取或创建限速器
        RateLimiter rateLimiter = rateLimiterRegistry.rateLimiter("user");

        //获取度量
        RateLimiter.Metrics metrics = rateLimiter.getMetrics();
        //获取估计剩下可用的调用数,一般用于测试和评估
        //如果有些调用还在进行,它可以为负值
        int availablePermissions = metrics.getAvailablePermissions();
        //获取估计当前还在运行的调用数,一般用于测试和评估
        int numberOfWaitingThreads = metrics.getNumberOfWaitingThreads();

        //事件的监听
        rateLimiter.getEventPublisher()
                //监听调用失败事件
                .onFailure(env->{
                    System.out.println("事件类型:["+env.getEventType().name()+"]");
                })
                //监听成功事件
                .onSuccess(env->{
                    System.out.println("事件类型:["+env.getEventType().name()+"]");
                })
                //监听成功或失败事件
                .onEvent(env->{
                    System.out.println("事监听所有事件,件类型:["+env.getEventType().name()+"]");
                });

        //动态修改 手动修改
        //在单位时间戳内限制的速率
//        rateLimiter.changeLimitForPeriod(50);
        //请求等待1秒,如果仍旧没有运行则将请求丢失
//        rateLimiter.changeTimeoutDuration(Duration.ofSeconds(1L));


        //定义事件,但是没有发送请求
        CheckedFunction0<UserInfo> decorateSupplier
                = RateLimiter.decorateCheckedSupplier(rateLimiter,()->reqUser(id));
        //发送请求
        Try<UserInfo> result = Try.of(decorateSupplier)
                //降级
                .recover(ex->{
                    ex.printStackTrace();
                    return new UserInfo(ERROR_ID,"",ex.getMessage());
                });
        return result.get();
    }
    //获取用户信息
    private UserInfo reqUser(Long id){
        String url = "http://eureka-client-user/user/info/{id}";
        System.out.println("获取用户"+ id);
        return restTemplate.getForObject(url,UserInfo.class,id);
    }

    @Override
    public UserInfo getUser3(Long id){
        //获取舱壁
        Bulkhead bulkhead = bulkheadRegistry.bulkhead("USER");

        //动态修改 手动操作
//        BulkheadConfig customCfg = BulkheadConfig.custom()
//                //请注意,最大等待数据是不允许动态修改的,因此这里的设置是无效的
//                .maxWaitDuration(Duration.ofSeconds(15))
//                //重置最大并发数
//                .maxConcurrentCalls(20)
//                .build();
//        bulkhead.changeConfig(customCfg);


        //获取度量
//        Bulkhead.Metrics metrics = bulkhead.getMetrics();
//        //获取可用并发数
//        metrics.getAvailableConcurrentCalls();



        //监听事件
        bulkhead.getEventPublisher()
                //调用许可事件
                .onCallPermitted(evt->{
                    System.out.println("调用许可事件");
                })
                //调用完成事件
                .onCallFinished(evt->{
                    System.out.println("调用完成事件");
                })
                //拒绝调用事件
                .onCallRejected(evt->{
                    System.out.println("拒绝调用事件");
                })
                //监听所有事件
                .onEvent(evt->{
                    System.out.println(evt.getEventType().name()+"监听所有事件");
                });


        //描述事件但没有发送
        CheckedFunction0<UserInfo> decoratedSupplier
                = Bulkhead.decorateCheckedSupplier(bulkhead,()->reqUser(id));
        //发送
        Try<UserInfo> result = Try.of(decoratedSupplier)
                //降级方法
                .recover(ex->{
                    ex.printStackTrace();
                    return new UserInfo(ERROR_ID,"",ex.getMessage());
                });

        return result.get();
    }


    //重试机制实现
    @Override
    public ResultMessage exp(){
        //获取重拾起
        Retry retry = retryRegistry.retry("timeout");
        
        //获取度量
//        Retry.Metrics metrics = retry.getMetrics();
//        //返回所有重试后的失败调用数
//        long numberOfFailedCallsWithRetryAttempt = metrics.getNumberOfFailedCallsWithRetryAttempt();
//        //返回为重试的失败调用数
//        long numberOfFailedCallsWithoutRetryAttempt = metrics.getNumberOfFailedCallsWithoutRetryAttempt();
//        //返回未重试后成功调用的次数
//        long numberOfSuccessfulCallsWithRetryAttempt = metrics.getNumberOfSuccessfulCallsWithRetryAttempt();
//        //返回重试后成功调用的次数
//        long numberOfSuccessfulCallsWithoutRetryAttempt = metrics.getNumberOfSuccessfulCallsWithoutRetryAttempt();


        //动态修改
        retryConfig = RetryConfig.custom()
                //最大尝试次数(默认Wie3次)
                .maxAttempts(5)
                //重试时间间隔(默认为500毫秒)
                .waitDuration(Duration.ofSeconds(1))
                //配置在何种异常下,放弃重试
                .ignoreExceptions(HttpServerErrorException.InternalServerError.class)
                //根据请求所得结果判断是否继续重试
                .retryOnResult(result->(result == null))
                .build();


        //得到事件发布者
        retry.getEventPublisher()
                //监听重试事件
                .onRetry(evt->{
                    System.out.println("重试");
                })
                //监听执行错误事件
                .onError(evt->{
                    System.out.println("执行错误");
                })
                //监听忽略错误事件
                .onIgnoredError(evt->{
                    System.out.println("忽略错误");
                })
                //监听执行成功事件
                .onSuccess(evt->{
                    System.out.println("执行成功");
                })
                //监听所有事件,根据事件类型判断是何种事件
                .onEvent(evt->{
                    System.out.println("事件类型名称:"+evt.getEventType().name());
                });


        //描述事件但没有发送
        CheckedFunction0<ResultMessage> decoratedSupplier
                = Retry.decorateCheckedSupplier(retry,()->reqExp());

        //发送请求
        Try<ResultMessage> result = Try.of(decoratedSupplier)
                //降级方法
                .recover(ex->new ResultMessage(false,"异常信息"+ex.getMessage()));

        return result.get();
    }

    private ResultMessage reqExp(){
        //不使用注入的restTemplate,原因是Ribbon存在重试机制
        RestTemplate restTmpl = new RestTemplate();
        String url = "http://localhost:6001/hystrix/exp/boot";
        return restTmpl.getForObject(url,ResultMessage.class);
    }


    /**
     * 获取Ehcache的一个缓存实例
     * @param id 用户编号
     * @return
     */
    private javax.cache.Cache<String,UserInfo> getCacheInstance(Long id){
        //获取缓存提供者,适合只有一个JCache实现的情况
        CachingProvider cachingProvider = Caching.getCachingProvider();
        /**
         * //如果系统有多种JCache,则根据具体实现类名获取缓存提供者 使用Ehcache缓存
         * CachingProvider cachingProvider = Caching.getCachingProvider("org.ehcache.jsr107.EhcacheCachingProvider");
         */
        //获取缓存管理器
        CacheManager cacheManager = cachingProvider.getCacheManager();
        //尝试获取名称为"user_"+id的缓存
        javax.cache.Cache<String,UserInfo> cacheInstance  = cacheManager.getCache("user_"+id);
        //获取失败,则创建缓存实例
        if(cacheInstance == null){
            //缓存配置类
            MutableConfiguration<String,UserInfo> config
                    = new MutableConfiguration<>();
            //设置缓存键值类型
            config.setTypes(String.class,UserInfo.class);
            //创建一个JCache对象,键值为"user_"+id
            cacheInstance = cacheManager.createCache("user_"+id,config);
        }
        return cacheInstance;
    }

    /**
     * 缓存过程实现
     * 对未命中的会进行自动缓存
     * @param id
     * @return
     */
    @Override
    public UserInfo cacheUserInfo(Long id){

        //获取名称为["user_"+id]的缓存
        javax.cache.Cache<String,UserInfo> cacheInstance = getCacheInstance(id);
        //通过Resilience4j的Cache捆绑JCache的缓存实例
        //此处的Cache类全限定名为io.github.resilience4j.cache.Cache
        //和getCacheInstance 方法的javax.cache.Cache不同
        Cache<String, UserInfo> cache = Cache.of(cacheInstance);

        //获取度量
//        Cache.Metrics metrics = cache.getMetrics();
//        //获取缓存命中的次数
//        long numberOfCacheHits = metrics.getNumberOfCacheHits();
//        //获取缓存未命中的次数
//        long numberOfCacheMisses = metrics.getNumberOfCacheMisses();

        //事件监听
        cache.getEventPublisher()
                //缓存命中的事件
                .onCacheHit(evt->{
                    System.out.println("命中");
                })
                //缓存未命中事件
                .onCacheMiss(event -> {
                    System.out.println("未命中");
                })
                //错误事件
                .onError(event -> {
                    System.out.println("出现错误");
                })
                //监听所有事件
                .onEvent(event -> {
                    System.out.println("事件类型名称:"+event.getEventType().name());
                });

        //描述事件
        CheckedFunction1<String,UserInfo> cacheFunction = Cache.decorateCheckedSupplier(cache,()->reqUser(id));
        //获取结果 ,现从缓存获取,键为["user_"+id],失败则从执行请求逻辑
        UserInfo user = Try.of(()->cacheFunction.apply("user_"+id)).get();
        return user;
    }


    /**
     * 时间限制器的使用
     * @return
     */
    @Override
    public ResultMessage timeout(){
        //创建时间限制器
        TimeLimiter timeLimiter = TimeLimiter.of(timeLimiterConfig);
        //采用单线程
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        //创建Supplier对象,并描述Supplier事件逻辑
        Supplier<Future<ResultMessage>>futureSupplier =
                //submit方法,提交任务并执行等待返回结果
                ()->executorService.submit(()->{
                    //不使用主人的RestTemplate ,因为不想启用Ribbon的超时机制
                    RestTemplate restTmpl = new RestTemplate();
                    String url = "http://eureka-client-user/hystrix/timeout";
                    return restTmpl.getForObject(url,ResultMessage.class);
                });
        //时间限制器捆绑事件
        Callable<ResultMessage> callable =
                TimeLimiter.decorateFutureSupplier(timeLimiter,futureSupplier);
        //获取结果
        Try<ResultMessage> result = Try.of(()->callable.call())
                //降级方法
                .recover(ex->new ResultMessage(false,"执行超时"));

        return result.get();
    }

    /**
     * 混合使用组件
     * @param id
     * @return
     */
    @Override
    public UserInfo mixUserInfo(Long id){

        //断路器
        CircuitBreaker circuitBreaker = circuitBreakerRegistry.circuitBreaker("user");
        //具体事件
        Callable<UserInfo> call = ()->reqUser(id);;
        //断路器捆绑事件
        Callable<UserInfo> call1 = CircuitBreaker.decorateCallable(circuitBreaker,call);
        //舱壁
        Bulkhead bulkhead = bulkheadRegistry.bulkhead("user");
        //舱壁捆绑断路器逻辑
        Callable<UserInfo> call2 = Bulkhead.decorateCallable(bulkhead,call1);
        //获取或创建限速器
        RateLimiter rateLimiter = rateLimiterRegistry.rateLimiter("user");
        //限速器捆绑舱壁事件
        Callable<UserInfo> call3 = RateLimiter.decorateCallable(rateLimiter,call2);
        //重试机制
        Retry retry = retryRegistry.retry("timeout");
        //重试捆绑事件
        Callable<UserInfo> call4 = Retry.decorateCallable(retry,call3);
        //获取名称为"user"+id的缓存实例
        javax.cache.Cache<String,UserInfo> cacheInstance = getCacheInstance(id);
        Cache<String,UserInfo> cache = Cache.of(cacheInstance);
        //缓存捆绑限速事件
        CheckedFunction1<String,UserInfo> cacheFunc = Cache.decorateCallable(cache,call4);
        //创建事件限时器
        TimeLimiter timeLimiter = TimeLimiter.of(timeLimiterConfig);
        //采用单线程池
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        //描述限时事件
        Supplier<Future<UserInfo>> supplier
                = ()->executorService.submit(()->{
                    UserInfo cacheResult = null;
                    try{
                        cacheResult = cacheFunc.apply("user_"+id);
                    }catch (Throwable e){
                        e.printStackTrace();
                    }
                    return cacheResult;
        });
        //限时器捆绑缓存事件
        Callable<UserInfo> calls = TimeLimiter.decorateFutureSupplier(timeLimiter,supplier);
        //获取结果
        Try<UserInfo> result =  Try.of(()->calls.call())
                //降级方法
                .recover(ex-> new UserInfo(ERROR_ID,"",ex.getMessage()));
        return result.get();

    }
}
